/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hylanda.processors.HNGetSnapShot;

import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.logging.ComponentLog;
import org.apache.commons.io.IOUtils;
import org.apache.http.ParseException;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.util.StandardValidators;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.hylanda.processors.HNGetSnapShot.util.GenUrlCrc64;
import com.hylanda.processors.HNGetSnapShot.util.HttpUtil;
import com.hylanda.processors.HNGetSnapShot.util.NifiUtil;

import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

@Tags({"haina,快照,snapshot"})
@CapabilityDescription("获取网页快照")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class HNGetSnapShot extends AbstractProcessor {

	public static final PropertyDescriptor URL_FIELD = new PropertyDescriptor.Builder().name("URL_FIELD")
			.displayName("url所在的字段").required(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
			.build();
    public static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder().name("API_URL")
			.displayName("API_URL").required(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
			.build();
    public static final PropertyDescriptor ENCODING = new PropertyDescriptor.Builder().name("ENCODING")
			.displayName("ENCODING").addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
			.build();
    public static final PropertyDescriptor JUST_STATUS = new PropertyDescriptor
    		.Builder().name("JUST_STATUS")
    		.displayName("仅检测数据是否存在快照")
    		.description("仅检测数据是否存在快照")
    		.required(false)
    		.allowableValues(new String[] { "true","false" })
    		.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
    		.build();
    public static final Relationship SUCCESS = new Relationship.Builder()
            .name("SUCCESS")
            .description("SUCCESS")
            .build();
    public static final Relationship FAILURE = new Relationship.Builder()
            .name("FAILURE")
            .description("FAILURE")
            .build();

    private List<PropertyDescriptor> descriptors;

    private Set<Relationship> relationships;

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
        descriptors.add(API_URL);
        descriptors.add(URL_FIELD);
        descriptors.add(JUST_STATUS);
        descriptors.add(ENCODING);
        this.descriptors = Collections.unmodifiableList(descriptors);

        final Set<Relationship> relationships = new HashSet<Relationship>();
        relationships.add(SUCCESS);
        relationships.add(FAILURE);
        this.relationships = Collections.unmodifiableSet(relationships);
    }

    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {

    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        FlowFile flowFile = session.get();
        if ( flowFile == null ) {
            return;
        }
       
        long startTime = System.currentTimeMillis();
        String api_url = null;
        try {
        	api_url =context.getProperty(API_URL).getValue().trim();
		} catch (Exception e) {
			// TODO: handle exception
		}
        String url_field = null;
        try {
        	url_field = context.getProperty(URL_FIELD).getValue().trim();
		} catch (Exception e) {
			// TODO: handle exception
		}
        String just_status = null;
        try {
        	just_status =context.getProperty(JUST_STATUS).getValue().trim();
		} catch (Exception e) {
			// TODO: handle exception
		}
       
        String encoding = "utf-8";
        try {
        	encoding =context.getProperty(ENCODING).getValue().trim();
		} catch (Exception e) {
			// TODO: handle exception
		}
        if (encoding==null||encoding.equals("")) {
        	encoding = "utf-8";
		}
        
        String content = NifiUtil.getContent(session, flowFile);
        
        ComponentLog log = getLogger();
      
        if (content==null) {
        	log.info("content is null");
        	session.transfer(flowFile, FAILURE);
			return;
		}
        //log.info("GetATags content :"+content);
        JSONObject jsonObject = JSONObject.parseObject(content);
        String url = jsonObject.getString(url_field);
        
        JSONObject queryJson = new JSONObject();
        queryJson.put("url", url);
        queryJson.put("type", 1);
        queryJson.put("check", "0");
        
        Map<String, String> map = new HashMap<>();
		map.put("appkey", "6044344985872289337");
		map.put("v", "2.0");
		map.put("format", "json");
		map.put("query", queryJson.toJSONString());
		
		JSONObject responseJson = new JSONObject();
		JSONObject resultJson = new JSONObject();
		String rusult_body = "";
        
        try {
			rusult_body = HttpUtil.send(api_url, map, encoding);
			responseJson = JSONObject.parseObject(rusult_body);
			resultJson = responseJson.getJSONObject("response");
		} catch (ParseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        if (resultJson.containsKey("resource")) { //获取到快照
        	
        	if (just_status!=null&&just_status.equals("true")) {
        		flowFile = session.write(flowFile, (in, out) -> {
        			try (OutputStream outputStream = new BufferedOutputStream(out)) {				
        				outputStream.write(IOUtils.toByteArray(jsonObject.toJSONString()));
        			}
        		});
			}else{
				JSONArray array = resultJson.getJSONArray("resource");
				if (array!=null&&array.size()>0) {
					JSONObject jsonObject2 = array.getJSONObject(0);
					flowFile = session.write(flowFile, (in, out) -> {
	        			try (OutputStream outputStream = new BufferedOutputStream(out)) {				
	        				outputStream.write(IOUtils.toByteArray(jsonObject2.toJSONString()));
	        			}
	        		});
				}else{
					flowFile = session.write(flowFile, (in, out) -> {
	        			try (OutputStream outputStream = new BufferedOutputStream(out)) {				
	        				outputStream.write(IOUtils.toByteArray(jsonObject.toJSONString()));
	        			}
	        		});
					session.transfer(flowFile, FAILURE);
		        	return;
				}
				
			}
        	
        	String session_id = flowFile.getAttribute(CoreAttributes.UUID.key());
            
            String uuid = GenUrlCrc64.GenCrc64(UUID.randomUUID().toString());
           // String session_id = flowFile.getAttribute(CoreAttributes.UUID.key());
            String instance_id = getIdentifier();
            JSONObject logJSON = new JSONObject();
            JSONObject msgJSON = new JSONObject();
            
            Date date = new Date();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String time = sdf.format(date);
            msgJSON.put("time", time);
            msgJSON.put("status", "success");
            
            logJSON.put("msg", msgJSON);
            logJSON.put("session_id", session_id);
            logJSON.put("uuid", uuid);
            logJSON.put("instance_id", instance_id);
            //getLogger().billing(session_id);
            getLogger().info(logJSON.toJSONString());
        	
        	
        	session.transfer(flowFile, SUCCESS);
        	return;
		}else{ //没有获取到快照快照
			flowFile = session.write(flowFile, (in, out) -> {
				try (OutputStream outputStream = new BufferedOutputStream(out)) {				
					outputStream.write(IOUtils.toByteArray(jsonObject.toJSONString()));
				}
			});
			session.transfer(flowFile, FAILURE);
			return;
		}
        
        //long endTime = System.currentTimeMillis();
        
        
        
        
        
        
    }
    
}
